package com.im.service.message.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.im.codec.pack.message.ChatMessageAck;
import com.im.codec.pack.message.MessageReceiveServerAckPack;
import com.im.service.config.AppConfig;
import com.im.common.constant.Constants;
import com.im.common.enums.ConversationTypeEnum;
import com.im.common.enums.command.MessageCommand;
import com.im.common.model.ClientInfo;
import com.im.common.model.RestResponse;
import com.im.common.model.message.MessageContent;
import com.im.common.model.message.OfflineMessageContent;
import com.im.service.message.model.req.SendMessageReq;
import com.im.service.message.model.resp.SendMessageResp;
import com.im.service.message.service.P2PMessageService;
import com.im.service.utils.CallbackService;
import com.im.service.utils.ConversationIdGenerate;
import com.im.service.utils.MessageProducer;
import com.im.service.utils.RedisSeq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 点对点消息服务类
 */
@Slf4j
@Service
public class P2PMessageServiceImpl implements P2PMessageService {

    @Autowired
    private CheckSendMessageServiceImpl checkSendMessageService; // 消息发送前的校验服务
    @Autowired
    private MessageProducer messageProducer; // 消息生产者
    @Autowired
    private MessageStoreServiceImpl messageStoreService; // 消息存储服务
    @Autowired
    private RedisSeq redisSeq; // Redis序列号生成器
    @Autowired
    private AppConfig appConfig; // 应用配置
    @Autowired
    private CallbackService callbackService; // 回调服务

    private final ThreadPoolExecutor threadPoolExecutor; // 线程池

    {
        /**
         * 创建一个线程池，包含8个核心线程和8个最大线程，线程空闲时间为60秒，
         * 等待队列容量为1000，使用LinkedBlockingDeque作为等待队列，线程工厂使用匿名内部类实现，
         * 通过AtomicInteger类保证线程名称的唯一性，线程为守护线程。
         */
        final AtomicInteger num = new AtomicInteger(0);
        threadPoolExecutor = new ThreadPoolExecutor(8, 8, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("message-process-thread-" + num.getAndIncrement());
                return thread;
            }
        });
    }

    /**
     * 处理消息
     *
     * @param messageContent 消息内容
     */
    public void process(MessageContent messageContent) {
        log.info("消息开始处理：{}", messageContent.getMessageId());
        // 判断是否是重复消息
        MessageContent messageFromMessageIdCache = messageStoreService.getMessageFromMessageIdCache(messageContent.getAppId(), messageContent.getMessageId(), MessageContent.class);
        if (messageFromMessageIdCache != null) {
            threadPoolExecutor.execute(() -> {
                // 1.回复发送方的消息确认
                ack(messageContent, RestResponse.success());
                // 2.同步消息给发送方的在线设备
                syncToSender(messageFromMessageIdCache, messageFromMessageIdCache);
                // 3.向接收方的在线设备发送消息
                List<ClientInfo> clientInfos = dispatchMessage(messageFromMessageIdCache);
                // 如果接收方没有设备在线
                if (clientInfos.isEmpty()) {
                    // 发送接收确认给发送方，要带上是服务端发送的标识
                    receiverAck(messageFromMessageIdCache);
                }
            });
            return;
        }

        // 如果开启了回调，执行回调操作
        RestResponse response = RestResponse.success();
        if (appConfig.isSendMessageAfterCallback()) {
            response = callbackService.beforeCallback(messageContent.getAppId(), Constants.CallbackCommand.SendMessageBefore, JSONObject.toJSONString(messageContent));
        }
        // 如果回调失败，不发送消息，发送ack
        if (!response.isStatus()) {
            ack(messageContent, response);
            return;
        }

        // 生成消息序列号
        long seq = redisSeq.doGetSeq(messageContent.getAppId() + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(messageContent.getFromId(), messageContent.getToId()));
        messageContent.setMessageSequence(seq);

        threadPoolExecutor.execute(() -> {
            // 存储消息
            messageStoreService.storeP2PMessage(messageContent);
            // 存储离线消息
            OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
            BeanUtils.copyProperties(messageContent, offlineMessageContent);
            offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
            messageStoreService.storeOfflineMessage(offlineMessageContent);
            // 插入数据
            // 1. 回复发送方的消息确认
            ack(messageContent, RestResponse.success());
            // 2. 同步消息给发送方的在线设备
            syncToSender(messageContent, messageContent);
            // 3. 向接收方的在线设备发送消息
            List<ClientInfo> clientInfos = dispatchMessage(messageContent);
            // 将消息存入缓存
            messageStoreService.setMessageFromMessageIdCache(messageContent.getAppId(), messageContent.getMessageId(), messageContent);
            if (clientInfos.isEmpty()) {
                // 发送接收确认给发送方，要带上是服务端发送的标识
                receiverAck(messageContent);
            }
            // 执行后续回调操作
            if (appConfig.isSendMessageAfterCallback()) {
                callbackService.callback(messageContent.getAppId(), Constants.CallbackCommand.SendMessageAfter, JSONObject.toJSONString(messageContent));
            }
            log.info("消息处理完成：{}", messageContent.getMessageId());
        });
    }


    /**
     * 分发消息给在线客户端
     *
     * @param messageContent 消息内容
     * @return 在线客户端列表
     */
    public List<ClientInfo> dispatchMessage(MessageContent messageContent) {
        List<ClientInfo> clientInfos = messageProducer.sendToUser(messageContent.getToId(), MessageCommand.MSG_P2P, messageContent, messageContent.getAppId());
        return clientInfos;
    }

    /**
     * 发送ack消息
     *
     * @param messageContent 消息内容
     * @param responseVO     响应结果
     */
    public void ack(MessageContent messageContent, RestResponse responseVO) {
        log.info("msg ack,msgId={},checkResult{}", messageContent.getMessageId(), responseVO.getCode());
        ChatMessageAck chatMessageAck = new ChatMessageAck(messageContent.getMessageId(), messageContent.getMessageSequence());
        responseVO.setData(chatMessageAck);
        // 发送消息
        messageProducer.sendToUser(messageContent.getFromId(), MessageCommand.MSG_ACK, responseVO, messageContent);
    }

    /**
     * 发送接收确认给发送方
     *
     * @param messageContent 消息内容
     */
    public void receiverAck(MessageContent messageContent) {
        // 创建接收确认消息
        MessageReceiveServerAckPack pack = new MessageReceiveServerAckPack();
        pack.setFromId(messageContent.getToId());
        pack.setToId(messageContent.getFromId());
        pack.setMessageKey(messageContent.getMessageKey());
        pack.setMessageSequence(messageContent.getMessageSequence());
        pack.setServerSend(true);
        // 发送接收确认消息给发送方
        messageProducer.sendToUser(messageContent.getFromId(), MessageCommand.MSG_RECIVE_ACK, pack, new ClientInfo(messageContent.getAppId(), messageContent.getClientType(), messageContent.getImei()));
    }

    /**
     * 同步消息给发送方在线客户端
     *
     * @param messageContent 消息内容
     * @param clientInfo     客户端信息
     */
    public void syncToSender(MessageContent messageContent, ClientInfo clientInfo) {
        messageProducer.sendToUserExceptClient(messageContent.getFromId(), MessageCommand.MSG_P2P, messageContent, messageContent);
    }

    /**
     * 检查发送方和接收方的权限
     *
     * @param fromId 发送方ID
     * @param toId   接收方ID
     * @param appId  应用ID
     * @return 响应结果
     */
    public void imServerPermissionCheck(String fromId, String toId, Integer appId) {
        // 检查发送者是否被禁言或屏蔽
        checkSendMessageService.checkSenderForvidAndMute(fromId, appId);
        // 检查好友关系是否正常
        checkSendMessageService.checkFriendShip(fromId, toId, appId);
    }

    /**
     * 发送消息
     *
     * @param req 请求参数
     * @return 发送结果
     */
    public SendMessageResp send(SendMessageReq req) {
        SendMessageResp sendMessageResp = new SendMessageResp();
        MessageContent message = new MessageContent();
        BeanUtils.copyProperties(req, message);
        // 插入数据
        messageStoreService.storeP2PMessage(message);
        sendMessageResp.setMessageKey(message.getMessageKey());
        sendMessageResp.setMessageTime(System.currentTimeMillis());
        // 发消息给同步在线端
        syncToSender(message, message);
        // 发消息给对方在线端
        dispatchMessage(message);
        return sendMessageResp;
    }
}
